package com.boot2.core.dao.batch.insert;

import com.boot2.core.utils.StringUtils;
import com.boot2.core.utils.ThreadPoolTaskExecutorUtil;
import lombok.Data;
import lombok.extern.apachecommons.CommonsLog;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.List;
import java.util.Map;

/**
 * @author zhangweilin
 * @Description: 批量提交数据载体
 * @date 2019/7/7
 */
@Data
@CommonsLog
public class BatchCommit<T> {

    private static final ThreadPoolTaskExecutor executor = ThreadPoolTaskExecutorUtil.createThreadPoolTaskExecutor(0.3);

    Object commitLock = new Object();

    /**
     * 数据集合的标识
     */
    private String key;

    /**
     * 每次批量最多提交的数量, 默认1000条,而如果积累到1000条的数据耗时较长，
     * 比如并发较低时，则会不等积累到1000条就会被触发，触发的条件为配置的定时间隔时间
     */
    private int batchMaxCommitSize = 1000;
    /**
     * 每次批量提交的间隔时间,单位毫秒，默认1000ms
     */
    private long intervalCommitTime = 1000;

    /**
     * 此值表示队列中积累到这个数量就立刻提交，与定时无关
     */
//    private int batchCommitSize = 1000;
//    private int batchCommitSize = 1000;

    /**
     * 队列长度一到指定值，就释放的锁,以达到自动批量提交目的
     */
//    private CyclicBarrier cyclicBarrier = new CyclicBarrier(batchCommitSize);

    /**
     * 默认情况，按基于内存的预提交
     */
    private BatchCommitProviderEnum provider = BatchCommitProviderEnum.memory;

    /**
     * 批量提交的数据载体实现
     */
    private Map<String, BatchCommitData> batchCommitDataMap;

    /**
     * 批量提交的Provider枚举
     */
    public enum BatchCommitProviderEnum {
        redis, memory
    }

    public BatchCommit(String key) {
        this.key = key;
    }

    public BatchCommitData getBatchCommitData() {
        String name = provider.name();
        name = StringUtils.changFirstWord(name, StringUtils.toUpperCase);
        return batchCommitDataMap.get("batchCommitDataWith" + name);
    }

    /**
     * 是否为空
     *
     * @return
     */
    public boolean isEmpty() {
        BatchCommitData batchCommitData = getBatchCommitData();
        return batchCommitData.isEmpty(key);
    }

    /**
     * 取长度
     * @return
     */
    public long size() {
        BatchCommitData batchCommitData = getBatchCommitData();
        return batchCommitData.size(key);
    }

    /**
     * 向队列添加数据
     *
     * @param t
     */
    @Async
    public void preCommit(T t) {

        BatchCommitData batchCommitData = getBatchCommitData();
        batchCommitData.preCommit(key, t);
//        taskQueue.offer(t);
//                cyclicBarrier.await();
        Long size = batchCommitData.size(key);
        if (size >= 10000) {
            if (size%5000==0) {
                log.error("【"+key+"】中待提交数据已超过10000条，当前已积累【"+size+"】条，请注意注意");
            }
        }
        if (size >= batchMaxCommitSize) {
            if (log.isDebugEnabled()) {
                log.debug("\n\n预提交,【"+key+"】,size:"+size+",batchMaxCommitSize: "+batchMaxCommitSize+", 超过临界值，准备唤醒批量提交");
            }
            synchronized (commitLock) {
                if (log.isDebugEnabled()) {
                    log.debug("预提交,【"+key+"】,唤醒批量提交2，进入synchronized");
                }
                commitLock.notifyAll();
                if (log.isDebugEnabled()) {
                    log.debug("预提交,【"+key+"】,唤醒批量提交成功");
                }
            }
        }
    }

    /**
     * 从队列取数据
     */
    public T poll() {
        BatchCommitData batchCommitData = getBatchCommitData();
        return batchCommitData.poll(key);
//        return taskQueue.poll();
    }

    /**
     * 从队列取数据
     */
    public List<T> pollList(int size) {
        BatchCommitData batchCommitData = getBatchCommitData();
        return batchCommitData.pollList(key, size);
    }

    public void getLock() {
        try {
            if (log.isDebugEnabled()) {
                log.debug("\n\n【"+key+"】,获取锁,等待被唤醒批量提交");
            }
            synchronized (commitLock) {
                if (log.isDebugEnabled()) {
                    log.debug("【"+key+"】,获取锁2,进入synchronized模块,等待被唤醒批量提交");
                }
                commitLock.wait(intervalCommitTime);
                if (log.isDebugEnabled()) {
                    log.debug("【"+key+"】,获取锁3, 被唤醒批量提交");
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            log.error("【"+key+"】,获取锁失败");
        }
    }


}
